package cn.iocoder.yudao.module.forward.mqtt;

import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttMessage;

@Slf4j
public abstract class MqttCallBackImpl implements IMqttCallBack {
    private MqttManager mMqttManager;
    protected MessageHandler mMessageHandler;
    protected String tag;

    public MqttCallBackImpl(MqttManager mqttManager, MessageHandler handler, String tag) {
        mMqttManager = mqttManager;
        mMessageHandler = handler;
        this.tag = tag;
    }

    @Override
    public void connectComplete(MqttInfo mqttInfo, boolean reconnect, String host) {
        if (reconnect) {
            log.info("{}({})重连成功: {}", tag, mqttInfo.getName(), host);
        } else {
            log.info("{}({})连接成功: {}", tag, mqttInfo.getName(), host);
        }
        mqttInfo.getTopics().clear();
        subscribeTopics(mqttInfo.getId());
    }

    @Override
    public void connectionLost(MqttInfo mqttInfo, Throwable throwable) {
        log.error("{}({})断开连接:{}, {}", tag, mqttInfo.getName(), mqttInfo, throwable);
        mqttInfo.getTopics().clear();
    }

    @Override
    public void deliveryComplete(MqttInfo mqttInfo, IMqttDeliveryToken iMqttDeliveryToken) {
        try {
            log.info("{}({})发送成功: msgId({})", tag, mqttInfo.getName(), iMqttDeliveryToken.getMessageId());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void messageArrived(MqttInfo mqttInfo, String topic, MqttMessage mqttMessage) throws Exception {
        try {
            String msg = null;
            if (mqttInfo.getCharset() == null) {
                msg = new String(mqttMessage.getPayload());
            } else {
                msg = new String(mqttMessage.getPayload(), mqttInfo.getCharset());
            }
            log.info("{}({})收到消息: topic({}), msg: {}", tag, mqttInfo.getName(), topic, msg);
            if (mMessageHandler != null) {
                mMessageHandler.handleMessage(mqttInfo, topic, msg);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void onSubscribe(MqttInfo mMqttInfo, boolean ret, String topic) {
        String msg = ret ? "成功" : "失败";
        log.info("{}({})订阅主题{}: {}", tag, mMqttInfo.getName(), msg, topic);
    }

    @Override
    public void onUnsubscribe(MqttInfo mMqttInfo, boolean ret, String topic) {
        String msg = ret ? "成功" : "失败";
        log.info("{}({})取消订阅主题{}: {}", tag, mMqttInfo.getName(), msg, topic);
    }

    public abstract void subscribeTopics(Integer id);
}
